package org.example.project.common.middleware.controller;

import org.example.project.common.contant.R;
import org.example.project.common.middleware.constant.ApiRequestMapping;
import org.example.project.common.middleware.service.KafkaService;
import org.example.project.common.middleware.service.dto.KafkaDTO;
import org.example.project.common.web.RestCrudController;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.validation.constraints.NotBlank;
import java.util.List;
import java.util.Map;

/**
 * @author wenxy
 * @date 2020/10/9
 */
@RestController
@RequestMapping(ApiRequestMapping.KAFKA_API)
@ConditionalOnClass(KafkaTemplate.class)
public class KafkaApiController implements RestCrudController<KafkaDTO, String> {

    @Autowired
    KafkaService defaultKafkaService;


    @Override
    public R<String> save(KafkaDTO kafkaDTO) {
        return R.success(defaultKafkaService.send(kafkaDTO));
    }

    @Override
    public R<KafkaDTO> getById(String s) {
        throw new UnsupportedOperationException();
    }

    @GetMapping("/partitions/{topic}")
    public R<List> partitionsFor(@NotBlank @PathVariable("topic") String s) {
        return R.success(defaultKafkaService.partitionsFor(s));
    }

    @GetMapping("/metrics")
    public R<Map> metrics() {
        return R.success(defaultKafkaService.metrics());
    }

}
